package com.afdor.rws.event.driven.channel.rocketmq.producer.topic;

import com.afdor.rws.annotation.extension.Extension;
import com.afdor.rws.event.driven.channel.producer.AbstractEventProducer;
import com.afdor.rws.event.driven.channel.producer.EventProducer;
import com.afdor.rws.event.driven.channel.rocketmq.config.RocketMqEventConfig;
import com.afdor.rws.event.driven.channel.rocketmq.utils.RocketMqUtils;
import com.afdor.rws.event.driven.entities.Event;
import com.afdor.rws.event.driven.entities.holder.CallbackEvent;
import com.afdor.rws.event.driven.entities.holder.ResultEvent;
import lombok.Getter;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.SendStatus;
import org.apache.rocketmq.client.producer.TransactionListener;
import org.apache.rocketmq.client.producer.TransactionMQProducer;
import org.apache.rocketmq.client.producer.TransactionSendResult;
import org.apache.rocketmq.common.message.Message;
import org.apache.rocketmq.remoting.common.RemotingHelper;

import java.io.UnsupportedEncodingException;
import java.util.List;
import java.util.Objects;
import java.util.concurrent.LinkedBlockingQueue;
import java.util.concurrent.ThreadFactory;
import java.util.concurrent.ThreadPoolExecutor;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicInteger;

/**
 * RocketMq生产者实现
 *
 * @author 悭梵
 * @date Created in 2018-12-18 14:41
 */
@Slf4j
@Extension(value = "rocketMqTransactionTopic", order = Extension.NORMAL_ORDER - 100, scope = Extension.Scope.PROTOTYPE, interfaceClass = EventProducer.class)
public class RocketMqTransactionEventProducer extends AbstractEventProducer {

    @Getter
    private TransactionMQProducer producer;

    @Getter
    private final TransactionListener transactionListener;

    public RocketMqTransactionEventProducer(TransactionListener transactionListener) {
        this.transactionListener = transactionListener;
    }

    @Override
    public <T extends EventProducer> T init() {
        super.init();
        if (this.producer != null) {
            throw new RuntimeException("已执行初始化操作，请不要再重复执行初始化操作了。");
        }

        try {
            RocketMqEventConfig config = getConfig().getEventConfig();
            this.producer = new TransactionMQProducer(RocketMqUtils.getTopicName(getEventSourceName()));
            this.producer.setDefaultTopicQueueNums(config.getDefaultTopicQueueNums());
            this.producer.setNamesrvAddr(config.getNamesrvAddr());
            this.producer.setSendMessageWithVIPChannel(Boolean.FALSE);
            this.producer.setSendMsgTimeout(1000);
            this.producer.setExecutorService(new ThreadPoolExecutor(config.getCorePoolSize(), config.getMaxPoolSize(), 100, TimeUnit.SECONDS, new LinkedBlockingQueue<>(config.getQueueSize()), new ThreadFactory() {
                /**
                 * 线程计数器
                 */
                private AtomicInteger threadIndex = new AtomicInteger(0);

                /**
                 * 线程组
                 */
                private ThreadGroup threadGroup = new ThreadGroup("rocket-mq-producer-client-thread");

                @Override
                public Thread newThread(Runnable r) {
                    Thread thread = new Thread(threadGroup, r, config.getThreadNamePrefix() + threadIndex.getAndIncrement());
                    thread.setPriority(Thread.MAX_PRIORITY);
                    return thread;
                }
            }, new ThreadPoolExecutor.CallerRunsPolicy()));
            this.producer.setTransactionListener(this.transactionListener);
            this.producer.start();
        } catch (MQClientException e) {
            throw new RuntimeException("生产者初始化失败", e);
        }
        return (T) this;
    }

    @Override
    public void sendAsyncEvent(Event event) {
        try {
            Message msg = new Message(RocketMqUtils.getTopicName(getEventSourceName()), null, event.getId(), ((String) getSerialization().serialize(event)).getBytes(RemotingHelper.DEFAULT_CHARSET));
            TransactionSendResult sendResult = getProducer().sendMessageInTransaction(msg, event.getId());
            if (!Objects.equals(SendStatus.SEND_OK, sendResult.getSendStatus())) {
                throw new RuntimeException("消息发送失败：" + sendResult);
            }
        } catch (MQClientException | UnsupportedEncodingException e) {
            throw new RuntimeException("消息发送失败", e);
        }
    }

    @Override
    public void batchSendAsyncEvent(List<Event> eventList) {
        for (Event event : eventList) {
            sendAsyncEvent(event);
        }
    }

    @Override
    public void asyncSendEvent(CallbackEvent callbackEvent) {
        throw new RuntimeException("RocketMq MQ事物消息不支持同步回调消息发送");
    }

    @Override
    public <V> ResultEvent<V> sendEvent(Event event) {
        throw new RuntimeException("RocketMq MQ事物消息不支持同步回调消息发送");
    }

    @Override
    public void destroy() {
        super.destroy();
        if (this.producer != null) {
            this.producer.shutdown();
            this.producer = null;
        }
    }
}
